package com.lfsenior.topic;

import jdk.nashorn.internal.ir.SetSplitState;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.Charset;

/**
 * Created by dlupan on 2018/10/31
 */
public class TopicSender {
    // TCP 地址
    public static final String BROKER_URL="tcp://localhost:61616";
    // 目标
    public static final String DESTINATION="TOPIC-QUEUE";

    /**
     * 发送消息
     * @param session
     * @param publisher
     */
    public static void sendMesssage(TopicSession session, TopicPublisher publisher) throws Exception{
        for(int i=0;i<5;i++){
            String message = "发送消息第" + (i + 1) + "条";
            MapMessage map = session.createMapMessage();
            map.setString("text",message);
            map.setLong("time",System.currentTimeMillis());

            System.out.println(map+"1");
            publisher.send(map);
        }
        Message message = session.createMessage();
    }

    public static void run() throws Exception{
        // 创建连接工厂
        TopicConnectionFactory factory = new ActiveMQConnectionFactory(BROKER_URL);
        // 创建连接
        TopicConnection connection = factory.createTopicConnection();
        // 启动连接
        connection.start();
        // 创建一个session会话
        TopicSession session = connection.createTopicSession(true, Session.AUTO_ACKNOWLEDGE);
        // 创建一个消息主题
        Topic topic = session.createTopic(DESTINATION);
        // 创建发布者
        TopicPublisher publisher = session.createPublisher(topic);
        // 设置持久化模式
        publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        sendMesssage(session,publisher);

        // 提交会话
        session.commit();

        // 关闭连接
        session.close();
        connection.close();
    }

    public static void main(String[] args) throws Exception{
        TopicSender.run();
    }

    public static String toString(final InputStream input, final Charset encoding) throws IOException {
        StringBuilder sb = new StringBuilder();
        InputStreamReader in = new InputStreamReader(input, encoding);
        char[] BUFFER = new char[1024];
        long count=0;
        int n;
        if(-1!=(n=in.read(BUFFER))){
            sb.append(BUFFER,0,n);
            count+=n;
        }
        return sb.toString();
    }

}
